package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime220815.bean.ExamQuestionAnswerUserBean;
import com.atguigu.edu.realtime220815.bean.PlayVideoTimesBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import jdk.internal.org.objectweb.asm.tree.analysis.Value;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.planner.expressions.In;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/19 1:57
 **/
public class DwsExamQuestionAnswerUserNumWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_exam_question_detail";
        String groupId = "DwsExamQuestionAnswerUserNumWindow";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic,groupId);
        DataStreamSource<String> kfkDS = env.addSource(flinkKafkaConsumer);
//{"id":"495965","exam_id":"33308","paper_id":"188","question_id":"744","user_id":"2408","is_correct":"0","score":"0.0","date_id":"2023-02-18","ts":"1676703783"}

        SingleOutputStreamOperator<JSONObject> jsonObjDs = kfkDS.map(JSON::parseObject);

        KeyedStream<JSONObject, String> keyDS = jsonObjDs.keyBy(value -> value.getString("question_id"));
        SingleOutputStreamOperator<ExamQuestionAnswerUserBean> processDS = keyDS.process(
                new KeyedProcessFunction<String, JSONObject, ExamQuestionAnswerUserBean>() {
                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDes = new ValueStateDescriptor<>("valueStateDes", String.class);
                        valueState = getRuntimeContext().getState(valueStateDes);
                    }

                    @Override
                    public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, ExamQuestionAnswerUserBean>.Context ctx, Collector<ExamQuestionAnswerUserBean> out) throws Exception {
                        String question_id = value.getString("question_id");
                        String user_id = value.getString("user_id");
                        String sign = "corrected";
                        String is_correct = value.getString("is_correct");
                        Long ts = value.getLong("ts");
                        Long answer_user_count = 0L;
                        Long correct_user_count = 0L;
                        String uidValue = valueState.value();
                        if(StringUtils.isNotEmpty(user_id)) {
                            if (StringUtils.isEmpty(uidValue)) {
                                answer_user_count = 1L;
                                valueState.update(user_id);
                                uidValue=valueState.value();
                            }

                            if (uidValue.equals(user_id) && "1".equals(is_correct)) {
                                correct_user_count = 1L;
                                valueState.update(sign);
                                uidValue=valueState.value();
                            }
                            ExamQuestionAnswerUserBean bean = ExamQuestionAnswerUserBean.builder()
                                    .questionId(question_id)
                                    .answer_user_count(answer_user_count)
                                    .correct_user_count(correct_user_count)
                                    .ts(ts)
                                    .build();
                            out.collect(bean);
                        }

                    }
                }
        );
        //processDS.print();

        SingleOutputStreamOperator<ExamQuestionAnswerUserBean> withWatermarkDS = processDS.assignTimestampsAndWatermarks(WatermarkStrategy.<ExamQuestionAnswerUserBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<ExamQuestionAnswerUserBean>() {
                    @Override
                    public long extractTimestamp(ExamQuestionAnswerUserBean element, long recordTimestamp) {
                        return element.getTs()*1000;
                    }
                }));

        KeyedStream<ExamQuestionAnswerUserBean, String> keyKeyDS = withWatermarkDS.keyBy(value -> value.getQuestionId());

        WindowedStream<ExamQuestionAnswerUserBean, String, TimeWindow> windowDS = keyKeyDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<ExamQuestionAnswerUserBean> reduceDS = windowDS.reduce(
                new ReduceFunction<ExamQuestionAnswerUserBean>() {
                    @Override
                    public ExamQuestionAnswerUserBean reduce(ExamQuestionAnswerUserBean value1, ExamQuestionAnswerUserBean value2) throws Exception {
                        value1.setCorrect_user_count(value1.getCorrect_user_count() + value2.getCorrect_user_count());
                        value1.setAnswer_user_count(value1.getAnswer_user_count() + value2.getAnswer_user_count());
                        return value1;
                    }
                },
                new WindowFunction<ExamQuestionAnswerUserBean, ExamQuestionAnswerUserBean, String, TimeWindow>() {

                    @Override
                    public void apply(String s, TimeWindow window, Iterable<ExamQuestionAnswerUserBean> input, Collector<ExamQuestionAnswerUserBean> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        long now = System.currentTimeMillis();

                        for (ExamQuestionAnswerUserBean examQuestionAnswerUserBean : input) {
                            examQuestionAnswerUserBean.setStt(DateFormatUtil.toYmdHms(start));
                            examQuestionAnswerUserBean.setEdt(DateFormatUtil.toYmdHms(end));
                            examQuestionAnswerUserBean.setTs(now);
                            out.collect(examQuestionAnswerUserBean);
                        }
                    }
                }
        );

        SingleOutputStreamOperator<ExamQuestionAnswerUserBean> withQuestionNameDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<ExamQuestionAnswerUserBean>("dim_test_question_info") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, ExamQuestionAnswerUserBean obj) {
                        String question_name = dimInfoJsonObj.getString("question_txt".toUpperCase());
                        obj.setQuestionName(question_name);
                    }

                    @Override

                    public String getKey(ExamQuestionAnswerUserBean obj) {
                        return obj.getQuestionId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        withQuestionNameDS.print();
        withQuestionNameDS.addSink(ClickHouseUtils.getClickHouseSinkFunction(
                "insert into dws_exam_question_answer_user_num_window values(?,?,?,?,?,?,?)"
        ));

        env.execute();
    }
}
